package com.yykj.drup.server.message;

import com.google.common.reflect.TypeToken;
import com.yykj.common.util.GsonUtil;
import com.yykj.drup.server.service.OrderInfoService;
import com.yykj.drup.server.service.OrderService;
import com.yykj.message.center.client.dto.TDMQMessage;
import com.yykj.message.center.client.dto.pay.DrugOrderCancelMessageBodyDTO;
import com.yykj.message.center.client.dto.pay.PayMessageBodyDTO;
import com.yykj.message.center.client.enums.OrderType;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class PayServiceMessageService {

    @Autowired
    private OrderService orderService;
    @Autowired
    private OrderInfoService orderInfoService;


    //从配置文件中获取 serviceURL 接入地址、Token 密钥、Topic 全名和 Subscription 名称（均可从控制台复制）
    @Value("${tdmq.serviceUrl}")
    private String serviceUrl;
    @Value("${tdmq.token}")
    private String token;

    @Value("${tdmq.namespace}")
    private String namespace;

    private String prefix = "persistent://pulsar-pjnbmw52podb/";

    private PulsarClient client;

    private List<Consumer<byte[]>> consumerList = new ArrayList<>();

    private Consumer<byte[]> consumer;

    MessageListener messageListener;

    @PostConstruct
    public void init() throws Exception {
        messageListener = new MessageListener() {
            @Override
            public void received(Consumer consumer, Message message) {
                try {
                    if (onMessage(message)) {
                        consumer.acknowledge(message);
                    } else {
                        consumer.negativeAcknowledge(message);
                    }
                } catch (PulsarClientException e) {
                    consumer.negativeAcknowledge(message);
                    e.printStackTrace();
                }
            }
        };

        client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .authentication(AuthenticationFactory.token(token))
                .build();
        System.out.println(">> pulsar client created.");
        //创建订单支付通知-消费者
        consumer = client.newConsumer()
                .topic(prefix + namespace + "/pay-success")
                .subscriptionName("drug-client")
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(messageListener)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();
        consumer = client.newConsumer()
                .topic(prefix + namespace + "/drug-cancel-order")
                .subscriptionName("drug-client")
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(messageListener)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();
        consumerList.add(consumer);

    }

    @PreDestroy
    public void destroy() throws Exception {
        for (Consumer<byte[]> consumer : consumerList) {
            if (consumer != null) {
                consumer.close();
            }
            if (client != null) {
                client.close();
            }
        }

    }

    /**
     * 这里写消费业务逻辑
     *
     * @param message
     * @return return true: 消息ack  return false: 消息nack
     * @throws Exception 消息nack
     */
    private boolean onMessage(Message<byte[]> message) {
        //业务逻辑,延时类操作
        String msg = new String(message.getValue());
        log.info("Drug-Service -- 收到消息：{}", msg);

        try {
            TDMQMessage tdmqMessage = GsonUtil.fromJson(new String(message.getValue()),
                    new TypeToken<TDMQMessage>(){}.getType());
            switch (tdmqMessage.getType()){
                case PAY_SUCCESS:
                    log.info("Drug-Service -- 订单支付成功,开始处理");
                    TDMQMessage tdmqMessagePay = GsonUtil.fromJson(new String(message.getValue()),
                            new TypeToken<TDMQMessage<PayMessageBodyDTO>>(){}.getType());
                    PayMessageBodyDTO dto = (PayMessageBodyDTO)tdmqMessagePay.getData();
                    if (dto.getOrderType().equals(OrderType.drug)){
                        String orderNum = dto.getOrderId();
                        Integer status = orderService.updateOrderStatus(orderNum);
                        log.info("Drug-Service -- 订单支付成功,结束状态--" + status);
                        log.info("Drug-Service -- 订单支付成功,发起微分销下单--"+ orderNum);
                        //根据总订单编号执行药店微分销下单
                        orderInfoService.weifenxiaoPlaceOrder(orderNum);
                    }else {
                        log.info("Drug-Service -- 跳过不需要处理的订单 ");
                    }
                    break;
                case DRUG_ORDER_CANCEL:
                    log.info("Drug-Service -- 订单取消消息,开始处理");
                    TDMQMessage tdmqMessageCancel = GsonUtil.fromJson(new String(message.getValue()),
                            new TypeToken<TDMQMessage<DrugOrderCancelMessageBodyDTO>>(){}.getType());
                    DrugOrderCancelMessageBodyDTO cancelDto = (DrugOrderCancelMessageBodyDTO)tdmqMessageCancel.getData();
                    log.info("Drug-Service -- 订单取消消息,订单号--" + cancelDto.getOrderNum());
                    orderService.cancelOrderStatus(cancelDto.getOrderNum());
                    break;
                default:
                    break;
            }
            log.info("Drug-Service -- 消息处理 success：{}", msg);
            return true;
        } catch (Exception exception) {
            exception.printStackTrace();
            log.error("药品服务 -- 消息处理 failed：{}", msg);
        }
        return false;
    }
}
